from mrjob.job import MRJob, MRStep

import datetime


class UserValue(MRJob):

    def mapper_step1(self, _, line):
        lineinfo = line.strip().split(',')
        if len(lineinfo) == 6:
            if lineinfo[3] == 'buy':
                final_day = datetime.datetime.strptime('2017-12-03', "%Y-%m-%d").date()
                now_day = datetime.datetime.strptime(lineinfo[4], "%Y-%m-%d").date()
                if 0 <= (final_day - now_day).days <= 1:
                    r_value = 4
                elif 2 <= (final_day - now_day).days <= 3:
                    r_value = 3
                elif 4 <= (final_day - now_day).days <= 5:
                    r_value = 4
                elif 6 <= (final_day - now_day).days <= 8:
                    r_value = 1
                buy_cnt = 1
            else:
                r_value = 1
                buy_cnt = 0

            yield lineinfo[0], (r_value, buy_cnt)

    def reducer_step1(self, key, values):
        R = 0
        buy_sum = 0
        user_rfm = '挽留客户'
        for (r_value, buy_cnt) in values:
            if r_value > R:
                R = r_value

            buy_sum += buy_cnt

        if buy_sum >= 30:
            F = 4
        elif 20 <= buy_sum < 30:
            F = 3
        elif 10 <= buy_sum < 20:
            F = 2
        elif 0 <= buy_sum < 10:
            F = 1

        if R > 2 and F > 2:
            user_rfm = '价值客户'
        elif R <= 2 and F > 2:
            user_rfm = '保持客户'
        elif R > 2 and F <= 2:
            user_rfm = '发展客户'
        else:
            user_rfm = '挽留客户'

        yield user_rfm, None

    def mapper_step2(self, user_frm, _):
        yield user_frm, 1

    def reducer_step2(self, key, values):
        yield key,sum(values)

    def steps(self):
        return [
            MRStep(mapper=self.mapper_step1,
                   reducer=self.reducer_step1),

            MRStep(mapper=self.mapper_step2,
                   reducer=self.reducer_step2)
        ]

if __name__ == '__main__':
    UserValue.run()

